﻿
Imports System.Threading
Imports std
Imports remotecall
Imports centrosphere.gravitation

Namespace gravitation
    Namespace remotecallName
        Public Module _remotecallName
            Public Const pop As String = "pop"
            Public Const update As String = "update"
        End Module
    End Namespace
End Namespace

Public MustInherit Class gravitation_star
    Implements iconsumer

    Public Sub start(ByVal sourceQueueName As String) Implements iconsumer.start
    End Sub

    Protected MustOverride Sub start(ByVal port As Int32, ByVal connectionCount As Int64)
    Public MustOverride Sub [stop]() Implements iconsumer.stop

    Public Shared Function initial(ByVal name As String,
                                   ByVal config As configParser,
                                   ByVal gs As gravitation_star) As Boolean
        assert(Not gs Is Nothing)
        Dim port As Int32 = 0
        Dim connectionCount As Int64 = 0
        If read_port_connectionCount(config, name, port, connectionCount, 128) Then
            gs.start(port, connectionCount)
            Return True
        Else
            Return False
        End If
    End Function
End Class

Public MustInherit Class gravitation_planet
    Inherits producer
    Protected MustOverride Overloads Sub start(ByVal host As String,
                                               ByVal port As Int32,
                                               ByVal connectionCount As Int64,
                                               ByVal planetQueueName As bytesString)
    Protected MustOverride Overloads Sub start(ByVal host As String,
                                               ByVal port As Int32,
                                               ByVal connectionCount As UInt16)

    Public Shared Shadows Function initial(ByVal name As String,
                                           ByVal config As configParser,
                                           ByVal gp As gravitation_planet) As Boolean
        assert(Not gp Is Nothing)
        If producer.initial(name, config, gp) Then
            Dim host As String = Nothing
            Dim port As Int32 = 0
            Dim connectionCount As Int64 = 0
            If config.getparameter(name, "host", host) AndAlso
               read_port_connectionCount(config, name, port, connectionCount, Environment.ProcessorCount << 1) Then
                Dim planetQueueName As String = Nothing
                If config.getparameter(name, "planetQueueName", planetQueueName) Then
                    gp.start(host, port, connectionCount, New bytesString(planetQueueName))
                Else
                    gp.start(host, port, connectionCount)
                End If
                Return True
            Else
                raiseError(errorHandle.errorType.user, "no host / port has been set in ", name)
                Return False
            End If
        Else
            Return False
        End If
    End Function
End Class

'the hook up between planet and star
Public MustInherit Class gravitation_star(Of SERVER_T, INPUT_T, OUTPUT_T)
    Inherits gravitation_star

    Private server As SERVER_T = Nothing

    Protected MustOverride Function create_server(ByVal port As Int32, ByVal connectionCount As Int64) As SERVER_T
    Protected MustOverride Function start_server(ByVal server As SERVER_T,
                                                 ByVal port As Int32,
                                                 ByVal connectionCount As Int64) As Boolean
    Protected MustOverride Sub stop_server(ByVal server As SERVER_T)
    Protected MustOverride Sub register(ByVal server As SERVER_T,
                                        ByVal name As String,
                                        ByVal callback As _do(Of INPUT_T, pointer(Of OUTPUT_T), eventComb))
    Protected MustOverride Function parse_input_valid(ByVal input As INPUT_T) As Boolean
    Protected MustOverride Function parse_input_queuename(ByVal input As INPUT_T, ByRef o As bytesString) As Boolean
    Protected MustOverride Function parse_input_jobId(ByVal input As INPUT_T, ByRef o As bytesInt64) As Boolean
    Protected MustOverride Function parse_input_ijob(ByVal input As INPUT_T, ByRef o As ijob) As Boolean
    Protected MustOverride Function create_output(ByVal field_count As Int32) As OUTPUT_T
    Protected MustOverride Sub inject_output_result(ByVal output As OUTPUT_T, ByVal result As fromBytes)
    Protected MustOverride Sub inject_output_jobId(ByVal output As OUTPUT_T, ByVal index As fromBytes)
    Protected MustOverride Sub inject_output_ijob(ByVal output As OUTPUT_T, ByVal job As ijob)

    Protected NotOverridable Overrides Sub start(ByVal port As Int32, ByVal connectionCount As Int64)
        server = create_server(port, connectionCount)
        assert(Not server Is Nothing)
        register(server, remotecallName.pop, AddressOf pop)
        register(server, remotecallName.update, AddressOf update)
        If start_server(server, port, connectionCount) Then
            raiseError("gravitation_star started on port ", port)
        Else
            raiseError(errorHandle.errorType.system,
                       "cannot start rcserve4 on port ", port, ", ignore gravitation_star.")
        End If
    End Sub

    Private Function pop(ByRef arg As INPUT_T, ByRef prtn As pointer(Of OUTPUT_T)) As eventComb
        Dim bsQueueName As bytesString = Nothing
        Dim rtn As OUTPUT_T = Nothing
        Dim job As indexedJob = Nothing
        Dim queueName As String = Nothing
        If parse_input_queuename(arg, bsQueueName) AndAlso
           eva(queueName, +bsQueueName) AndAlso
           Not isemptystring(queueName) AndAlso
           jobqueues.at(queueName).pop(job) Then
            assert(Not job Is Nothing)
            rtn = create_output(3)
            inject_output_result(rtn, New bytesBoolean(True))
            inject_output_jobId(rtn, New bytesInt64(job.index()))
            inject_output_ijob(rtn, job.job())
        Else
            rtn = create_output(1)
            inject_output_result(rtn, New bytesBoolean(False))
        End If
        eva(prtn, rtn)
        Return Nothing
    End Function

    Private Function update(ByRef arg As INPUT_T, ByRef prtn As pointer(Of OUTPUT_T)) As eventComb
        Dim job As ijob = Nothing
        Dim bsQueueName As bytesString = Nothing
        Dim queueName As String = Nothing
        Dim bsJobId As bytesInt64 = Nothing
        Dim jobId As Int64 = 0
        Dim rtn As OUTPUT_T = Nothing
        rtn = create_output(1)
        If parse_input_queuename(arg, bsQueueName) AndAlso
           eva(queueName, +bsQueueName) AndAlso
           Not isemptystring(queueName) AndAlso
           parse_input_jobId(arg, bsJobId) AndAlso
           eva(jobId, +bsJobId) AndAlso
           parse_input_ijob(arg, job) AndAlso
           jobqueues.at(queueName).update(jobId,
                                          Sub(ByRef i As indexedJob)
                                              assert(i.job.copy(job))
                                              i.finish()
                                          End Sub) Then
            inject_output_result(rtn, New bytesBoolean(True))
        Else
            inject_output_result(rtn, New bytesBoolean(False))
        End If
        eva(prtn, rtn)
        Return Nothing
    End Function

    Public NotOverridable Overrides Sub [stop]()
        stop_server(server)
    End Sub
End Class

Public MustInherit Class gravitation_planet(Of CLIENT_T, RETURN_T)
    Inherits gravitation_planet
    Private stopped As Boolean = False
    Private connectionCount As Int32 = 0

    Private Shared ReadOnly waittimeMs As Int64 = 32
    Private Shared ReadOnly planetQueueName As bytesString = Nothing
    Private QPS As Int64 = 0
    Private LATENCY As Int64 = 0
    Private INTERNAL_LATENCY As Int64 = 0
    Private IDLE_PERCENTAGE As Int64 = 0

    Shared Sub New()
        planetQueueName = New bytesString(share.planetQueueName())
    End Sub

    Private Class target_t
        Public sourceIndex As Int64
        Public job As ijob = Nothing
        Public indexedJob As indexedJob = Nothing
    End Class

    Public Overrides Sub [stop]()
        Dim waitSeconds As Int32 = 10
        stopped = True
        While connectionCount > 0 AndAlso waitSeconds > 0
            sleepSeconds(1)
            waitSeconds -= 1
        End While
    End Sub

    Protected MustOverride Function create_client(ByVal host As String, ByVal port As Int32) As CLIENT_T
    Protected MustOverride Function invoke(ByVal client As CLIENT_T,
                                           ByVal name As String,
                                           ByRef ec As eventComb,
                                           ByVal rtn As pointer(Of RETURN_T),
                                           ByVal ParamArray paras() As toBytes) As Boolean
    Protected MustOverride Function parse_return_valid(ByVal r As RETURN_T) As Boolean
    Protected MustOverride Function parse_return_success(ByVal r As RETURN_T, ByRef o As bytesBoolean) As Boolean
    Protected MustOverride Function parse_return_sourceIndex(ByVal r As RETURN_T, ByRef o As bytesInt64) As Boolean
    Protected MustOverride Function parse_return_job(ByVal r As RETURN_T, ByRef o As ijob) As Boolean

    Private Sub startSingle(ByVal client As CLIENT_T,
                            ByVal planetQueueName As bytesString,
                            ByVal connection_index As Int64)
        Dim waittimeMs As Int64 = gravitation_planet(Of CLIENT_T, RETURN_T).waittimeMs * (connection_index + 1)
        Dim rcec As eventComb = Nothing
        Dim prtns As pointer(Of RETURN_T) = Nothing
        prtns = New pointer(Of RETURN_T)()
        Dim target As target_t = Nothing
        target = New target_t()
        Interlocked.Increment(Me.connectionCount)
        Dim [stop] As _do(Of Boolean) =
            Function() As Boolean
                Interlocked.Decrement(Me.connectionCount)
                Return goto_end()
            End Function
        Dim [step] As _do(Of _do(Of Boolean), _do(Of Boolean)) =
            Function(ByRef i As _do(Of Boolean)) As _do(Of Boolean)
                assert(Not i Is Nothing)
                Dim d As _do(Of Boolean) = Nothing
                d = i
                Return Function() As Boolean
                           If stopped Then
                               Return [stop]()
                           Else
                               Return d()
                           End If
                       End Function
            End Function
        Dim lastTotalMs As Int64 = 0
        Dim lastInternalMs As Int64 = 0
        assert_begin(New eventComb(
                            [step](Function() As Boolean
                                       lastTotalMs = Now.milliseconds()
                                       Return invoke(client,
                                                     remotecallName.pop,
                                                     rcec,
                                                     prtns,
                                                     planetQueueName) AndAlso
                                              waitfor(rcec) AndAlso
                                              goto_next()
                                   End Function),
                            [step](Function() As Boolean
                                       Dim rtns As RETURN_T = Nothing
                                       rtns = +prtns
                                       If rcec.end_result() Then
                                           If parse_return_valid(rtns) Then
                                               Dim suc As bytesBoolean = Nothing
                                               If parse_return_success(rtns, suc) AndAlso +suc Then
                                                   counter.increase(IDLE_PERCENTAGE, 0)
                                                   Dim index As bytesInt64 = Nothing
                                                   Dim job As ijob = Nothing
                                                   If parse_return_sourceIndex(rtns, index) AndAlso
                                                      parse_return_job(rtns, job) Then
                                                       target.sourceIndex = +index
                                                       target.job = job
                                                       counter.increase(QPS)
                                                       lastInternalMs = Now.milliseconds()
                                                       Return ijobStep(target.job,
                                                                       AddressOf goto_next,
                                                                       target.indexedJob)()
                                                   Else
                                                       raiseError(errorHandle.errorType.warning,
                                                                  "not a valid data received from gravitation_star ",
                                                                  "when status is true")
                                                       Return [goto](0)
                                                   End If
                                               Else
                                                   counter.increase(IDLE_PERCENTAGE, 100)
                                                   Return (Now.milliseconds() - lastTotalMs >= waittimeMs OrElse
                                                           waitfor(waittimeMs)) AndAlso
                                                          [goto](0)
                                               End If
                                           Else
                                               raiseError(errorHandle.errorType.warning,
                                                          "not a valid data received from gravitation_star")
                                               Return [goto](0)
                                           End If
                                       Else
                                           raiseError(errorHandle.errorType.system,
                                                      "failed to call ", remotecallName.pop)
                                           Return [goto](0)
                                       End If
                                   End Function),
                            [step](Function() As Boolean
                                       assert(target.indexedJob.finished() OrElse target.indexedJob.canceled())
                                       counter.recordTimeMs(INTERNAL_LATENCY, lastInternalMs)
                                       Return invoke(client,
                                                     remotecallName.update,
                                                     rcec,
                                                     prtns,
                                                     planetQueueName,
                                                     New bytesInt64(target.sourceIndex),
                                                     target.job) AndAlso
                                              waitfor(rcec) AndAlso
                                              goto_next()
                                   End Function),
                            [step](Function() As Boolean
                                       If rcec.end_result() Then
                                           counter.recordTimeMs(LATENCY, lastTotalMs)
                                           Return [goto](0)
                                       Else
                                           raiseError(errorHandle.errorType.system,
                                                      "failed to call ", remotecallName.update)
                                           Return goto_prev()
                                       End If
                                   End Function)))
    End Sub

    Protected NotOverridable Overrides Sub start(ByVal host As String,
                                                 ByVal port As Int32,
                                                 ByVal connectionCount As Int64,
                                                 ByVal planetQueueName As bytesString)
        Trace.Assert(QPS = 0)
        Trace.Assert(LATENCY = 0)
        Trace.Assert(INTERNAL_LATENCY = 0)
        Dim counterName As _do(Of String, String) =
            Function(ByRef s As String) As String
                Return strcat("GRAVITATION_PLANET_",
                              strtoupper(s), "_",
                              strtoupper(host), "_",
                              strtoupper(port), "_",
                              planetQueueName)
            End Function
        QPS = counter.registerRateAndLastRate(counterName("qps"))
        LATENCY = counter.registerAverageAndLastAverage(counterName("latency_ms"))
        INTERNAL_LATENCY = counter.registerAverageAndLastAverage(counterName("internal_latency_ms"))
        IDLE_PERCENTAGE = counter.registerAverageAndLastAverage(counterName("idle_percentage"))

        stopped = False
        Dim client As CLIENT_T = Nothing
        client = create_client(host, port)
        assert(Not client Is Nothing)

        If connectionCount <= 0 Then
            raiseError(errorHandle.errorType.user, "connectionCount should be > 0")
            connectionCount = 1
        End If

        While _dec(connectionCount) >= 0
            startSingle(client, planetQueueName, connectionCount)
        End While
    End Sub

    Protected NotOverridable Overrides Sub start(ByVal host As String,
                                                 ByVal port As Int32,
                                                 ByVal connectionCount As UInt16)
        start(host, port, connectionCount, planetQueueName)
    End Sub
End Class
